package cloud.xiguapi.ubas.analysis.market.model;

import cloud.xiguapi.ubas.model.MarketingUserBehavior;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 10:21
 * desc:
 */
public class SimulatedMarketingUserBehaviorSource implements SourceFunction<MarketingUserBehavior> {

    /**
     * 定义用户行为的集合
     */
    private final List<String> BEHAVIORS = Arrays.asList("CLICK", "DOWNLOAD", "INSTALL", "UNINSTALL");
    /**
     * 定义渠道的集合
     */
    private final List<String> CHANNELS = Arrays.asList("app store", "wechat", "qq", "weibo");
    /**
     * 定义随机数生成器
     */
    private final Random random = new Random();
    /**
     * 控制是否正常运行的标志位
     */
    private boolean running = true;

    private int count = 1024 * 2;

    @Override
    public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
        while (count >= 0 && running) {
            // 随机生成所有字段
            Long userId = random.nextLong();
            String behavior = BEHAVIORS.get(random.nextInt(BEHAVIORS.size()));
            String channel = CHANNELS.get(random.nextInt(CHANNELS.size()));
            Long timestamp = System.currentTimeMillis();

            // 发出数据
            ctx.collect(MarketingUserBehavior.builder()
                    .userId(userId)
                    .behavior(behavior)
                    .channel(channel)
                    .timestamp(timestamp)
                    .build());
            count--;
        }
    }

    @Override
    public void cancel() {
        // 停止运行
        this.running = false;
    }
}
